package org.qee.mq.custom;

import lombok.Data;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;

@Data
public class Topic {

    //目前为单机
    private static final ConcurrentHashMap<String, Topic> topicHashMap = new ConcurrentHashMap<>();


    private String topicName;

    private List<Consumer<?>> consumerList;

    private List<Producer> producerList;

    private Queue queue;

    public static String parentPath = "/Users/zhuqi/Documents/workspace/learning2/learning/week13";


    public Topic(String topicName) {
        this.topicName = topicName;
        this.consumerList = new ArrayList<>();
        this.producerList = new ArrayList<>();
        this.queue = new DefaultQueue(topicName, parentPath);
    }

    private void addProducer(Producer producer) {
        synchronized (this) {
            producerList.add(producer);
        }
    }

    private void addConsumer(Consumer<?> consumer) {
        synchronized (this) {
            consumerList.add(consumer);
        }
    }

    public static Topic register(String topicName, Producer producer) {
        if (topicHashMap.containsKey(topicName)) {
            return topicHashMap.get(topicName);
        }
        topicHashMap.computeIfAbsent(topicName, Topic::new);
        Topic topic = topicHashMap.get(topicName);
        topic.addProducer(producer);
        return topic;
    }

    public static Topic subcribe(String topicName, Consumer<?> consumer) {
        if (topicHashMap.containsKey(topicName)) {
            return topicHashMap.get(topicName);
        }
        topicHashMap.computeIfAbsent(topicName, Topic::new);
        Topic topic = topicHashMap.get(topicName);
        topic.addConsumer(consumer);
        return topic;
    }


    public boolean send(Message message) {
        return queue.offer(message);
    }

    public Queue.TakeResult poll(long offset, int size) throws IOException {
        return queue.take(offset, size);
    }

    public static void restart(String topicName) {
        Topic topic = new Topic(topicName);
        topicHashMap.put(topicName, topic);
        topic.reload();

    }

    private void reload() {
        queue.reload();

    }


}
